package cn.itcast.flink.tmall;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousEventTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.Comparator;
import java.util.PriorityQueue;
import java.util.Queue;
/**
* Flink 实现：模拟简易双11实时统计大屏
* - 实时计算出当天零点截止到当前时间的销售总额
* - 计算出销售top3类别
* - 每秒钟更新一次统计结果
*/
public class TmallBigScreen {
public static void main(String[] args) throws Exception {
            // 1. 执行环境-env
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            // 第一、设置时间语义：事件时间EventTime
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
            // 2. 数据源-source
            DataStreamSource<TmallOrder> orderStream = env.addSource(new OrderSource());
            //orderStream.printToErr();
            // 3. 数据转换-transformation
            // 第二、设置事件时间字段及watermark水位线
            SingleOutputStreamOperator<TmallOrder> timeStream = orderStream.assignTimestampsAndWatermarks(
            // 设置最大允许乱序或延迟数据为5秒
            new BoundedOutOfOrdernessTimestampExtractor<TmallOrder>(Time.seconds(5)) {
            FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss:SSS");
            @Override
            public long extractTimestamp(TmallOrder order) {
            long eventTime = System.currentTimeMillis() - 5 * 1000 ;
                try{
                    eventTime = format.parse(order.getOrderTime()).getTime();
                }catch (Exception e){ e.printStackTrace(); }
                return eventTime;
            } }
            );
//            timeStream.printToErr();
        // TODO：step1. 每秒统计今日各个类别销售额（window size：1d，trigger interval： 1s，keyBy：category）
            SingleOutputStreamOperator<CategoryAmount> categoryWindowStream = timeStream
        // a. 设置分组字段：类别category
                    .keyBy("category")
        // b. 事件时间窗口：1 day
                    .window(TumblingEventTimeWindows.of(Time.days(1)))
        // c. 设置触发器trigger：1 second
                    .trigger(ContinuousEventTimeTrigger.of(Time.seconds(1)))
        // d. 窗口聚合操作，实现AggregateFunction，其中金额转换BigDecimal
                    .aggregate(
        // TODO: 定义聚合函数， 对数据进行增量聚合操作
                            new AggregateFunction<TmallOrder, BigDecimal, Double>() { @Override
                            public BigDecimal createAccumulator() {
        // 初始化中间临时变量，此处创建BigDecimal对象
                                return new BigDecimal(0);
                            }@Override
                            public BigDecimal add(TmallOrder order, BigDecimal accumulator) {
        // 获取订单金额
                                Double orderAmount = order.getOrderAmount();
        // 累加操作
                                BigDecimal addBigDecimal = accumulator.add(new BigDecimal(orderAmount));
        // 返回值
                                return addBigDecimal;
                            }@Override
                            public BigDecimal merge(BigDecimal a, BigDecimal b) {
                                return a.add(b);
                            }@Override
                            public Double getResult(BigDecimal accumulator) {
                                return accumulator.setScale(2, RoundingMode.HALF_UP).doubleValue();
                            }
                            }, // TODO: 定义窗口函数， 对窗口数据进行计算并输出
                            new WindowFunction<Double, CategoryAmount, Tuple, TimeWindow>() {
                                FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");
                                @Override
                                public void apply(Tuple tuple,TimeWindow window,Iterable<Double> input,
                                                  Collector<CategoryAmount> out) throws Exception {
        // 类别category
                                    String category = ((Tuple1<String>)tuple).f0 ;
                                    // 窗口中消费金额
                                    Double windowAmount = input.iterator().next();
        // 窗口结束时间
                                    String computeDataTime = format.format(System.currentTimeMillis());
        // 输出结果
                                    out.collect(new CategoryAmount(category, windowAmount, computeDataTime));
                                } }
                    );
/*
            [2021-01-27 22:15:40]: 办公 = 120.84
            [2021-01-27 22:15:40]: 男装 = 415.24
            [2021-01-27 22:15:40]: 运动 = 319.08
            [2021-01-27 22:15:40]: 乐器 = 165.33
            [2021-01-27 22:15:40]: 户外 = 54.12
            [2021-01-27 22:15:40]: 家电 = 111.76
            [2021-01-27 22:15:40]: 图书 = 309.51
            [2021-01-27 22:15:40]: 游戏 = 92.55
            [2021-01-27 22:15:40]: 家具 = 199.87
            [2021-01-27 22:15:40]: 美妆 = 164.76
*/
//    categoryWindowStream.printToErr();
    // TODO: step2. 每秒钟统计消费额Top3类别和总销售额（window：1s，processTime 处理时间）
    /*
    - 销售总额 ; - Top3销售的类别 ; - 每秒计算一次输出
    */
        SingleOutputStreamOperator<String> resultStream = categoryWindowStream
    // 按照计算时间
                .keyBy("computeDateTime")
    // 设置窗口：1s
                .window(TumblingProcessingTimeWindows.of(Time.seconds(1)))
    // 窗口内聚合，获取总销售额和Top类别
                .apply(new WindowFunction<CategoryAmount, String, Tuple, TimeWindow>() {
                    @Override
                public void apply(Tuple tuple,
                                  TimeWindow window,
                                  Iterable<CategoryAmount> input,
                                  Collector<String> out) throws Exception {
    // TODO： a-1. 声明金额累加
                    BigDecimal sumDecimal = new BigDecimal(0.0) ;
    // TODO： a-2. 定义优先队列，存储CategoryAmount对象，按照金额降序排序
                    Queue<CategoryAmount> queue = new PriorityQueue<CategoryAmount>(
                            4, //
                            new Comparator<CategoryAmount>() { @Override
                            public int compare(CategoryAmount o1, CategoryAmount o2) {
                                int comp = 0;
                                if(o1.getTotalAmount() > o2.getTotalAmount()){comp = 1; // 升序
                                }else if(o1.getTotalAmount() < o2.getTotalAmount()){
                                    comp = -1 ; // 降序
                                }
                                return comp;
                            } } //
                    );
    // TODO： b. 遍历窗口数据，进行计算
                    for (CategoryAmount element : input) {
                        System.out.println("window: " + element);
    // b-1. 加入队列，获取Top3
                        queue.add(element);
                        if (queue.size() > 3) queue.poll();
    // b-2. 累加金额
                        sumDecimal = sumDecimal.add(new BigDecimal(element.getTotalAmount()));
                    }
    // TODO: c. 输出队列数据
                    String computeDataTime = ((Tuple1<String>) tuple).f0 ;
    // c-1. 输出总金额
                    double sumAmount = sumDecimal.setScale(2, RoundingMode.HALF_UP).doubleValue();
                    CategoryAmount all = new CategoryAmount("all", sumAmount, computeDataTime);
                    out.collect("All>>>>" + all.toString());
    // c.2. 输出Top3
                    StringBuilder builder = new StringBuilder(computeDataTime).append(", ");
                    for (CategoryAmount item : queue) {
                        builder.append(item.toContent()).append(", ");
                    }
                    String output = builder.toString();
                    out.collect("Top3>>>>" + output.substring(0, output.length() - 2));
            }
            });
        // 4. 数据终端-sink
//            resultStream.printToErr();
        /*
        window: [2021-01-27 22:31:07]: 家电 = 292.35
        window: [2021-01-27 22:31:07]: 办公 = 158.36
        window: [2021-01-27 22:31:07]: 男装 = 114.84
        window: [2021-01-27 22:31:07]: 图书 = 172.78
        window: [2021-01-27 22:31:07]: 家具 = 157.0
        window: [2021-01-27 22:31:07]: 乐器 = 304.14
        window: [2021-01-27 22:31:07]: 洗护 = 101.39
        window: [2021-01-27 22:31:07]: 运动 = 313.77
        window: [2021-01-27 22:31:07]: 游戏 = 22.7
        window: [2021-01-27 22:31:07]: 女装 = 143.94
        window: [2021-01-27 22:31:07]: 户外 = 45.32
        window: [2021-01-27 22:31:07]: 美妆 = 252.45
        All>>>>[2021-01-27 22:31:07]: all = 2079.04Top3>>>>2021-01-27 22:31:07, 家电 = 292.35, 乐器 = 304.14, 运动 = 313.77
        */
        // 5. 触发执行-execute
            env.execute(TmallBigScreen.class.getSimpleName());
} }